package com.wmh.baseservice.common.redis.mq.config;

import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.wmh.baseservice.common.redis.mq.core.RedisMqTemplate;
import com.wmh.baseservice.common.redis.mq.core.interceptor.RedisMessageInterceptor;
import com.wmh.baseservice.common.redis.mq.core.interceptor.RedisMessageInterceptorFactory;
import com.wmh.baseservice.common.redis.mq.core.job.RedisPendingMessageResendJob;
import com.wmh.baseservice.common.redis.mq.core.message.RedisMessageBase;
import com.wmh.baseservice.common.redis.mq.core.stream.AbstractRedisStreamMessage;
import com.wmh.baseservice.common.redis.mq.core.stream.AbstractRedisStreamMessageListener;
import com.wmh.baseservice.common.redis.mq.core.stream.RedisStreamFailHandler;
import com.wmh.baseservice.common.redis.mq.core.stream.RedisStreamFailHandlerFactory;
import com.wmh.baseservice.common.redis.serializer.KeyPrefixRedisSerializer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.List;
import java.util.Objects;
import java.util.Properties;

/**
 * 消息队列配置类
 * @author 谢骏
 * @date 2023/11/9 16:58
 **/

@Slf4j
@Configuration
public class RedisMqConfiguration {
    @Bean
    public RedisPendingMessageResendJob redisPendingMessageResendJob(RedissonClient redissonClient, RedisMqTemplate redisMqTemplate, RedisStreamFailHandlerFactory redisStreamFailHandlerFactory) {
        log.info("初始化 [RedisPendingMessageResendJob] 消息重发器");
        return new RedisPendingMessageResendJob(redissonClient, redisMqTemplate, redisStreamFailHandlerFactory);
    }

    @Bean
    public RedisMqTemplate redisMqTemplate(StringRedisTemplate stringRedisTemplate, RedisMessageInterceptorFactory redisMessageInterceptorFactory) {
        return new RedisMqTemplate(stringRedisTemplate, redisMessageInterceptorFactory);
    }

    /**
     * 注册消息拦截器工厂
     * 建立消息体与拦截器的关系
     */
    @Bean
    public RedisMessageInterceptorFactory redisMessageInterceptorFactory(List<RedisMessageInterceptor<? extends RedisMessageBase>> interceptors) {
        RedisMessageInterceptorFactory factory = new RedisMessageInterceptorFactory();
        interceptors.forEach(interceptor -> {
            factory.put(interceptor.getMessageClass(), (RedisMessageInterceptor<RedisMessageBase>) interceptor);
            log.info("[RedisMessageInterceptorFactory][注册 消息体类({}) 对应的拦截器 ({})]",
                    interceptor.getMessageClass(), interceptor.getClass());
        });
        return factory;
    }

    /**
     * 注册 stream 消息 失败处理器工厂
     * 建立消息体与拦截器之间的关系
     */
    @Bean
    @SneakyThrows
    public RedisStreamFailHandlerFactory redisStreamFailHandlerFactory(List<RedisStreamFailHandler<? extends AbstractRedisStreamMessage>> failHandlers) {
        RedisStreamFailHandlerFactory factory = new RedisStreamFailHandlerFactory();
        for (RedisStreamFailHandler<? extends AbstractRedisStreamMessage> handler : failHandlers) {
            factory.put(handler.getMessageClass().newInstance().getStreamKey(), handler);
            log.info("[RedisStreamFailHandler][注册 消息体类({}) 对应的失败处理器 ({})]",
                    handler.getMessageClass(), handler.getClass());
        }
        return factory;
    }

    /**
     * 注册 stream 消息监听器容器
     * 只有存在 监听 bean 的时候才需要注册 stream 监听
     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    @ConditionalOnBean(AbstractRedisStreamMessageListener.class)
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> recordStreamMessageListenerContainer(RedisMqTemplate redisMqTemplate, KeyPrefixRedisSerializer keyPrefixRedisSerializer,
                                                                                                                     List<AbstractRedisStreamMessageListener<?>> listeners) {
        // 校验版本号
        checkRedisVersion(redisMqTemplate.getStringRedisTemplate());
        // 创建监听器配置
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 拉取消息超时时间(间隔时间 1s 拉取一次)
                .pollTimeout(Duration.ofSeconds(1))
                // 一次最多拉取10条
                .batchSize(10)
                // 配置带项目健前缀的序列化
                .keySerializer(keyPrefixRedisSerializer)
                // 目标类型
                .targetType(String.class)
                .build();

        // 创建监听容器
        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer.create(
                Objects.requireNonNull(redisMqTemplate.getStringRedisTemplate().getConnectionFactory()), options);
        listeners.parallelStream().forEach(listener -> {
            log.info("[redisStreamMessageListenerContainer][开始注册 StreamKey({}) consumer({}) 对应的监听器({})]",
                    listener.getStreamKey(), listener.getConsumer(), listener.getClass().getName());
            // 设置 Consumer 消费进度，以最小消费进度为准
            StreamOffset<String> streamOffset = StreamOffset.create(listener.getStreamKey(), ReadOffset.lastConsumed());
            // 设置 Consumer 监听
            StreamMessageListenerContainer.StreamReadRequestBuilder<String> builder = StreamMessageListenerContainer.StreamReadRequest
                    .builder(streamOffset).consumer(listener.getConsumer())
                    .autoAcknowledge(false) // 不自动 ack
                    .cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false
            container.register(builder.build(), listener);
            log.info("[redisStreamMessageListenerContainer][完成注册 StreamKey({}) consumer({}) 对应的监听器({})]",
                    listener.getStreamKey(), listener.getConsumer(), listener.getClass().getName());
        });
        return container;
    }


    /**
     * 校验 Redis 版本号，是否满足最低的版本号要求！
     */
    private static void checkRedisVersion(StringRedisTemplate stringRedisTemplate) {
        // 获得 Redis 版本
        Properties info = stringRedisTemplate.execute((RedisCallback<Properties>) RedisServerCommands::info);
        String version = MapUtil.getStr(info, "redis_version");
        // 校验最低版本必须大于等于 5.0.0
        int majorVersion = Integer.parseInt(StrUtil.subBefore(version, '.', false));
        if (majorVersion < 5) {
            throw new IllegalStateException(StrUtil.format("您当前的 Redis 版本为 {}，小于最低要求的 5.0.0 版本！", version));
        }
    }
}
